package com.jiazhong.example.example.rocket.demo1.stock.consumer;

import com.alibaba.fastjson.JSONArray;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import com.jiazhong.example.example.rocket.demo1.commons.bean.Alipay;
import com.jiazhong.example.example.rocket.demo1.commons.bean.Stock;
import com.jiazhong.example.example.rocket.demo1.stock.service.StockService;
import jakarta.annotation.Resource;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.util.List;

@Component
@Slf4j
public class StockConsumer {
    @Resource
    private StockService stockService;

    @SneakyThrows
    @Scheduled(initialDelay = 1000)
    public void updateStock() {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("order_consumer_group");
        consumer.setNamesrvAddr("124.220.170.254:9876");
        consumer.setMessageModel(MessageModel.BROADCASTING);
        consumer.subscribe("Order", "alipay");
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {

                for (MessageExt msg : list) {
                    byte[] body = msg.getBody();
                    String json = new String(body, StandardCharsets.UTF_8);
                    Alipay alipay = JSONArray.parseObject(json, Alipay.class);
                    QueryWrapper<Stock> wrapper = new QueryWrapper<>();
                    wrapper.eq("book_id", alipay.getBookId());
                    Stock one = stockService.getOne(wrapper);

                    UpdateWrapper<Stock> updateWrapper = new UpdateWrapper<>();
                    updateWrapper.eq("book_id", alipay.getBookId());
                    updateWrapper.set("number", one.getNumber() - alipay.getNumber());
                    stockService.update(updateWrapper);
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        consumer.start();
    }
}
